package spring.cloud.service.task.service.impl;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.google.common.base.Strings;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.data.Stat;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import spring.cloud.service.task.domain.executor.Executor;
import spring.cloud.service.task.domain.executor.ExecutorStatus;
import spring.cloud.service.task.domain.executor.ExecutorType;
import spring.cloud.service.task.domain.task.ExecutableDateTimePointList;
import spring.cloud.service.task.domain.task.ZooKeeperTaskConfigurationTransferUtil;
import spring.cloud.service.task.domain.task_executor.TaskExecutor;
import spring.cloud.service.task.domain.task_executor.TaskExecutorStatus;
import spring.cloud.service.task.domain.task_executor.TaskExecutorStatusInfo;
import spring.cloud.service.task.domain.task_item.TaskItemExecuteInfo;
import spring.cloud.service.task.domain.task_item.TaskItemExecuteStatus;
import spring.cloud.service.task.dto.NamespaceDto;
import spring.cloud.service.task.dto.TaskDto;
import spring.cloud.service.task.mapper.NameSpaceDtoMapper;
import spring.cloud.service.task.mapper.TaskDtoMapper;
import spring.cloud.service.task.service.TaskService;
import spring.cloud.service.task.service.ZooKeeperService;
import spring.cloud.service.task.utils.PageInfo;
import spring.cloud.service.task.utils.Result;
import spring.cloud.tasks.common.*;
import spring.cloud.tasks.common.curator.Curator;
import spring.cloud.tasks.common.curator.CuratorFactory;
import spring.cloud.tasks.common.curator.CuratorInfo;
import spring.cloud.tasks.common.nodepaths.*;
import spring.cloud.tasks.common.task.TaskItemExecuteRenewalInfo;
import spring.cloud.tasks.common.task.TaskItemRenewalNodeInfo;

import javax.annotation.Resource;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.TimeUnit;


@Service
@Slf4j
public class TaskServiceImpl implements TaskService {
    public static int ONE_MINUTE = 1 * 60 * 1000;
    public static int TASK_CAN_BE_DELETE_TIME_LIMIT = 2 * ONE_MINUTE;// 任务可以被删除的时间限制(单位：ms)
    @Resource
    private ZooKeeperService zooKeeperService;
    @Resource
    private NameSpaceDtoMapper nameSpaceDtoMapper;
    @Resource
    private TaskDtoMapper taskDtoMapper;

    @Override
    public PageInfo getTaskList(String namespaceKey, String description, String status, Integer page, Integer limit) {
        int count = taskDtoMapper.getTaskListCount(namespaceKey, description, status);
        if (count == 0) {
            PageInfo pageInfo = new PageInfo(new ArrayList<spring.cloud.service.task.domain.zooKeeperAndNamespace.ZooKeeper>());
            return pageInfo;
        }
        PageInfo pageInfo = new PageInfo(page, limit, count);
        Integer pageOffset = pageInfo.getCurrentPageOffset();
        Integer pageSize = limit.intValue();
        List<TaskDto> taskDtoList = taskDtoMapper.getTaskList(namespaceKey, description, status, pageOffset, pageSize);
        //
        List<String> namespaceKeyList = new ArrayList<>();
        Map<String, String> namespaceKeyDescriptionToMap = new HashMap<>();
        taskDtoList.forEach(taskDto -> {
            if (!namespaceKeyList.contains(taskDto.getNamespaceKey())) {
                namespaceKeyList.add(taskDto.getNamespaceKey());
            }
        });
        namespaceKeyList.forEach((namespaceKeyValue) -> {
            NamespaceDto namespaceDto = nameSpaceDtoMapper.selectByNamespaceKey(namespaceKeyValue);
            if (namespaceDto != null) {
                namespaceKeyDescriptionToMap.put(namespaceKeyValue, namespaceDto.getDescription());
            }
        });
        taskDtoList.forEach(taskDto -> {
            if (namespaceKeyDescriptionToMap.containsKey(taskDto.getNamespaceKey())) {
                taskDto.setNamespaceDescription(namespaceKeyDescriptionToMap.get(taskDto.getNamespaceKey()));
            }
        });
        taskDtoList.forEach(taskDto -> {
            taskDto.setStatusFlag(YN.Y.equals(taskDto.getStatus()));
        });
        //
        pageInfo.setRecords(taskDtoList);
        return pageInfo;
    }


    @Transactional(rollbackFor = RuntimeException.class)
    @Override
    public Result addTask(TaskDto taskDto) {
        //参数校验
        Result result = validateTaskConfiguration(taskDto);
        if (result.getCode() != Result.success_code) {
            return result;
        }
        //DB校验
        NamespaceDto namespaceDto = nameSpaceDtoMapper.selectByNamespaceKey(taskDto.getNamespaceKey());
        if (namespaceDto == null) {
            return Result.fail("命名空间:" + JSON.toJSONString(taskDto.getNamespaceKey()) + "不存在");
        }
        int taskNo = 0;//根据命名空间获取最新的taskId
        taskDto.setTaskId(taskNo + "");
        String taskId = taskDto.getTaskId();
        TaskDto oldTaskDto = taskDtoMapper.getTaskDtoByNamespaceKeyAndTaskId(taskDto.getNamespaceKey(), taskId);
        if (oldTaskDto != null) {
            return Result.fail("命名空间:" + JSON.toJSONString(namespaceDto.getDescription()) + "中已经存在任务id" + JSON.toJSONString(taskDto.getTaskId()));
        }
        // 该域任务总数不能超过一定数量
        //指定命名空间多少任务 最多256.建议128。
        int taskNum = nameSpaceDtoMapper.getTaskNumByNamespaceKey(namespaceDto.getNamespaceKey());
        if (taskNum + 1 > namespaceDto.getTaskMaxNum()) {
            return Result.fail("命名空间:" + JSON.toJSONString(taskDto.getNamespaceKey()) + "任务数已经超过阀值:" + namespaceDto.getTaskMaxNum());
        }
        //
        //ZK校验
        String namespaceKey = taskDto.getNamespaceKey();
        //新生成一个四位的taskId
        CuratorInfo curatorInfo = zooKeeperService.getCuratorInfo(namespaceKey);
        Curator curator = CuratorFactory.getCurator(curatorInfo, CuratorFactory.shortTime_true);
        if (curator == null) {
            return Result.fail("连接ZooKeeper失败");
        }
        //这种情况概率很小 采用先DB事务先执行 ,ZK同时也是事务执行,如果ZK挂了 DB事务回滚。
        //如果真的出现了这种问题，那么新建一个taskId即可。有可能在编辑的时候是一个存在的taskId。
        //如果是因为点了两次造成 那么需要使用string.intern()进行同步控制，一把控制台是单机程序，且配置任务的为一个人。故这种问题很难出现，真的出现了则新建一个taskId+00x即可。
        if (curator.checkExists(TaskNodePaths.getTaskNodePath(taskId))) {
            return Result.fail(String.format("该任务(%s)在ZK已经存在", taskId));
        }
        // 添加该任务到数据库
        taskDto.setUpdateVersion(System.currentTimeMillis());
        taskDtoMapper.addTask(taskDto);
        // 添加该任务配置到zk
        ZooKeeperTaskConfigurationTransferUtil.createTaskConfigurationToZooKeeper(curator, taskDto);
        ZooKeeperTaskConfigurationTransferUtil.updateTaskUpdateVersion(curator);
        return Result.success("添加任务成功");

    }

    @Transactional(rollbackFor = RuntimeException.class)
    @Override
    public Result updateTask(TaskDto taskDto) {
        TaskDto oldTaskDto = taskDtoMapper.getTaskDtoByNamespaceKeyAndTaskId(taskDto.getNamespaceKey(), taskDto.getTaskId());
        if (oldTaskDto == null) {
            Result result = Result.fail(String.format("该任务(%s)不存在", taskDto.getTaskId()));
            return result;
        }
        //【从数据库拿出老的数据，与老的数据库中的该任务的配置对比，如果没有改变，则直接返回】这种方案存在问题 就是如果ZK和DB不一致 那么ZK永远更新不了。
        //if (JSON.toJSONString(oldTaskDto).equals(JSON.toJSONString(taskDto))){return Result.fail()};
        //add by huihui
        if (!oldTaskDto.getNamespaceKey().equals(taskDto.getNamespaceKey())) {
            Result result = Result.warn("危险:命名空间被修改");
            return result;
        }
        if (!oldTaskDto.getTaskType().equals(taskDto.getTaskType())) {
            Result result = Result.warn("危险:任务类型被修改");
            return result;
        }
        if (!oldTaskDto.getTaskId().equals(taskDto.getTaskId())) {
            Result result = Result.warn("危险:任务id被修改");
            return result;
        }
        //
        // 校验任务配置
        Result result = validateTaskConfiguration(taskDto);
        if (result.getCode() != Result.success_code) {
            return result;
        }
        //需要做一个处理，判读可执行时间点是否发生了改变.如果改变则需要生成一个新的签名
        String cronExpressionString = taskDto.getCronExpression();
        String pauseDatePeriod = taskDto.getPausePeriodDate();
        String pauseTimePeriod = taskDto.getPausePeriodTime();
        //String executableDateTimePointExpressionSignature = taskDto.getExecutableDateTimePointExpressionSignature();
        if (!oldTaskDto.getCronExpression().equals(cronExpressionString) ||//
                !oldTaskDto.getPausePeriodDate().equals(pauseDatePeriod) ||//
                !oldTaskDto.getPausePeriodTime().equals(pauseTimePeriod)//
        ) {
            taskDto.setFireDateTimeVersion(System.currentTimeMillis());
        }
        //ZK校验
        String namespaceKey = taskDto.getNamespaceKey();
        String taskId = taskDto.getTaskId();
        CuratorInfo curatorInfo = zooKeeperService.getCuratorInfo(namespaceKey);
        Curator curator = CuratorFactory.getCurator(curatorInfo, CuratorFactory.shortTime_true);
        if (!curator.checkExists(TaskNodePaths.getTaskNodePath(taskId))) {
            return Result.fail(String.format("该任务(%s)在ZK不存在", taskId));
        }
        // 更新该任务到数据库
        taskDto.setUpdateVersion(System.currentTimeMillis());
        taskDtoMapper.updateTask(taskDto);
        // 更新该任务配置到zk
        ZooKeeperTaskConfigurationTransferUtil.updateTaskConfigurationToZooKeeper(curator, taskDto);
        ZooKeeperTaskConfigurationTransferUtil.updateTaskUpdateVersion(curator);
        return Result.success("修改任务成功");
    }

    private Result validateTaskConfiguration(TaskDto taskDto) throws RuntimeException {
        // 任务类型必填
        if (StringUtils.isBlank(taskDto.getTaskType())) {
            return Result.fail("任务类型必填");
        }
        // 验证任务类型
        TaskType taskType = TaskType.getTaskType(taskDto.getTaskType());
        if (taskType == TaskType.UNKNOWN_TASK) {
            return Result.fail("任务类型未知");
        }
        // xid必填
        String xid = taskDto.getXid();
        if (StringUtils.isBlank(xid)) {
            return Result.fail("xid必填");
        }
        // 任务名必填
        String taskId = taskDto.getTaskId();
        if (StringUtils.isBlank(taskId)) {
            return Result.fail("taskId必填");
        }
        // 任务名只允许包含：数字0-9、小写字符a-z、大写字符A-Z、下划线_
        if (!taskId.matches("[0-9a-zA-Z_]*")) {
            return Result.fail("任务名只允许包含：数字0-9、小写字符a-z、大写字符A-Z、下划线_");
        }
        // 校验cron
        Result result = validateCronExpression(taskDto);
        if (result.getCode() != Result.success_code) {
            return result;
        }
        result = validateTime(taskDto);
        if (result.getCode() != Result.success_code) {
            return result;
        }
        try {
            result = validateTaskItem(taskDto);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            return Result.fail("解析任务分片数+任务分片序列号/任务分片参数信息映射+任务分片参数配置信息失败");
        }
        if (result.getCode() != Result.success_code) {
            return result;
        }
        if (taskType == TaskType.SIMPLE_SLICING) {
            //校验
        } else if (taskType == TaskType.TIMING_BASED_BATCH_PROCESSING ||//
                taskType == TaskType.USER_TIMING_BASED_BATCH_PROCESSING) {//
            //校验
        }
        return Result.success();
    }

    public Result validateCronExpression(TaskDto taskDto) {
        String cronExpressionString = taskDto.getCronExpression();
        String pauseDatePeriod = taskDto.getPausePeriodDate();
        String pauseTimePeriod = taskDto.getPausePeriodTime();
        long fireDateTimeVersion = taskDto.getFireDateTimeVersion();
        if (StringUtils.isBlank(cronExpressionString)) {
            return Result.fail("cron表达式不能为空");
        }
        DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        SuspendableCronExpression executableDateTimePointExpression = new SuspendableCronExpression(fireDateTimeVersion, cronExpressionString, pauseDatePeriod, pauseTimePeriod);
        ExecutableDateTimePointList executableDateTimePointList = new ExecutableDateTimePointList();
        executableDateTimePointList.setCronExpression(cronExpressionString);
        executableDateTimePointList.setNextFireDateTimeList(new ArrayList<String>());
        Date now = new Date();
        for (int i = 0; i < 10; i++) {
            Date next = executableDateTimePointExpression.getNextFireTimeAfter(now);
            if (next != null) {
                executableDateTimePointList.getNextFireDateTimeList().add(dateFormat.format(next));
                now = next;
            }
        }
        if (executableDateTimePointList.getNextFireDateTimeList().isEmpty()) {
            return Result.fail(String.format("cron表达式(%s)下次执行时间列表为空", cronExpressionString));
        }
        executableDateTimePointExpression.setVersion(System.currentTimeMillis());//new
        return Result.success();
    }

    public Result validateTime(TaskDto taskDto) {
        Integer preAllotMinutes = taskDto.getPreAllotMinutes();
        Integer timeoutMinutes = taskDto.getTimeoutMinutes();
        Integer timeoutAlarmMinutes = taskDto.getTimeoutAlarmMinutes();
        if (preAllotMinutes == null) {
            return Result.fail("预分配分钟数为空");
        }
        if (preAllotMinutes < 1) {
            return Result.fail("预分配分钟数小于1");
        }
        //
        if (timeoutMinutes == null) {
            return Result.fail("超时分钟数为空");
        }
        if ((timeoutMinutes == -1)) {
            //给一个足够长的时间进行超时处理 比如7天。
        } else {
            if (timeoutMinutes < 6) {
                return Result.fail("超时分钟数小于6");
            }
        }
        //如果timeoutMinutes==null或者==-1时,只是告警
        if (timeoutAlarmMinutes == null) {
            return Result.fail("超时告警分钟数为空");
        }
        if (timeoutAlarmMinutes < 6) {
            return Result.fail("超时告警分钟数小于6");
        }
        return Result.success();
    }

    //任务分片数
    //任务分片序列号和任务分片参数信息映射 格式如下:[{"taskItemIndex":1,taskItemParameter:{}},{"taskItemIndex":2,taskItemParameter:{}}]
    //任务分片参数配置信息
    private Result validateTaskItem(TaskDto taskDto) throws RuntimeException {
        String taskItemInfo = taskDto.getTaskItemInfo();
        if (taskItemInfo == null) {
            return Result.fail("任务stage信息为空");
        }
        //只有简单分片才允许配置分片参数信息。如果启动一个新的简单分片任务那么老的任务则需要停止。
        // 如果是批次任务则采用分布式任务小批量数据处理模式。
        if (taskDto.getTaskType().equals(TaskType.SIMPLE_SLICING)) {
            SimpleSlicingTaskTaskItemCountAndLoadAndParameterInfo simpleSlicingTaskTaskItemCountAndLoadAndParameterInfo = JSON.parseObject(taskItemInfo, SimpleSlicingTaskTaskItemCountAndLoadAndParameterInfo.class);
            try {
                //[{"taskItemIndex":1,taskItemParameter:{}},{"taskItemIndex":2,taskItemParameter:{}}]
                String taskItemIndexAndTaskItemParametersString = simpleSlicingTaskTaskItemCountAndLoadAndParameterInfo.getTaskItemIndexAndTaskItemParametersString();
                JSONArray jsonArray = JSON.parseArray(taskItemIndexAndTaskItemParametersString);
                if ((jsonArray.size() != simpleSlicingTaskTaskItemCountAndLoadAndParameterInfo.getTaskItemCount())) {//
                    return Result.fail("分片参数和分片总数不相等");
                }
                Map<String, String> taskItemIndexToTaskItemParameterMap = new HashMap<>();
                for (Object o : jsonArray) {
                    JSONObject jsonObject = (JSONObject) o;
                    int taskItemIndex = jsonObject.getIntValue("taskItemIndex");
                    String taskItemParameter = jsonObject.getString("taskItemParameter");
                    taskItemIndexToTaskItemParameterMap.put(String.valueOf(taskItemIndex), taskItemParameter.trim());//去掉空格
                }
                for (int i = 1; i <= simpleSlicingTaskTaskItemCountAndLoadAndParameterInfo.getTaskItemCount(); i++) {
                    if (!taskItemIndexToTaskItemParameterMap.containsKey(String.valueOf(i))) {
                        return Result.fail("任务分片index:" + i + "对应的分片参数信息不存在");
                    }
                }
            } catch (Exception e) {
                return Result.fail("解析任务分片序列号和任务分片参数信息映射");
            }
        } else {

            List<TaskStageInfo> taskStageAndTaskItemTypeAndCountAndLoadList = JSON.parseArray(taskItemInfo, TaskStageInfo.class);
            int size = taskStageAndTaskItemTypeAndCountAndLoadList.size();
            if (size == 0) {
                return Result.fail("任务stage信息为空");
            }
            Map<String, String> stageNumMap = new HashMap<>();
            for (int i = 1; i <= size; i++) {
                stageNumMap.put(String.valueOf(i), "");
            }
            for (TaskStageInfo taskStageAndTaskItemTypeAndCountAndLoad : taskStageAndTaskItemTypeAndCountAndLoadList) {
                String stage = taskStageAndTaskItemTypeAndCountAndLoad.stage;
                int stageNum = taskStageAndTaskItemTypeAndCountAndLoad.stageNum;
                if (stage == null) {
                    return Result.fail("任务Stage信息中存在stage字段为空的数据");
                }
                if (stageNum <= 0) {
                    return Result.fail("任务Stage信息中存在stage:" + stage + ",stageNum小于等于0");
                }
                if (!stageNumMap.containsKey(String.valueOf(stageNum))) {
                    return Result.fail("任务Stage信息中存在stage:" + stage + ",stageNum错误");
                }
                List<TaskStageInfo.TaskItemInfo> taskItemTypeAndCountAndLoadList = taskStageAndTaskItemTypeAndCountAndLoad.taskItemInfoList;
                if (taskItemTypeAndCountAndLoadList == null || taskItemTypeAndCountAndLoadList.size() == 0) {
                    return Result.fail("任务Stage信息中存在stage:" + stage + ",分片信息为空");
                }
                //
                for (TaskStageInfo.TaskItemInfo taskItemTypeAndCountAndLoad : taskItemTypeAndCountAndLoadList) {
                    String type = taskItemTypeAndCountAndLoad.taskItemType;
                    int taskItemCount = taskItemTypeAndCountAndLoad.taskItemCount;
                    int taskItemLoad = taskItemTypeAndCountAndLoad.taskItemLoad;
                    if (type == null) {
                        return Result.fail("任务Stage信息中存在stage:" + stage + ",分片信息中任务处理类型为空的数据");
                    }
                    if (taskItemCount < 1) {
                        return Result.fail("任务Stage信息中存在stage:" + stage + ",分片信息中任务处理类型为" + type + ",分片数小于1的数据");
                    }
                    if (taskItemLoad < 1) {
                        return Result.fail("任务Stage信息中存在stage:" + stage + ",分片信息中任务处理类型为" + type + ",分片权重小于1的数据");
                    }
                }
                // 分片参数不能小于分片总数
            }
        }

        return Result.success();
    }


    @Transactional(rollbackFor = RuntimeException.class)
    @Override
    public Result deleteTask(TaskDto taskDto) throws RuntimeException {
        String nameSpace = taskDto.getNamespaceKey();
        String taskId = taskDto.getTaskId();
        TaskDto oldTaskId = taskDtoMapper.getTaskDtoByNamespaceKeyAndTaskId(nameSpace, taskId);
        if (oldTaskId == null) {
            return Result.fail("不能删除该任务（" + taskId + "），因为该任务不存在");
        }
        CuratorInfo curatorInfo = zooKeeperService.getCuratorInfo(nameSpace);
        Curator curator = CuratorFactory.getCurator(curatorInfo, CuratorFactory.shortTime_true);
        String taskNodePath = TaskNodePaths.getTaskNodePath(taskId);
        boolean taskExists = curator.checkExists(taskNodePath);
        if (taskExists) {
            Stat stat = curator.getStat(taskNodePath);
            if (stat != null) {
                if (System.currentTimeMillis() - stat.getCtime() < TASK_CAN_BE_DELETE_TIME_LIMIT) {
                    return Result.fail(String.format("不能删除该任务(%s)，因为该任务创建时间距离现在不超过%d分钟", taskId, TASK_CAN_BE_DELETE_TIME_LIMIT / ONE_MINUTE));
                }
            }
            //ZK删除
            try {
                removeTaskFromZooKeeper(curator, taskId);
            } catch (Exception e) {
                return Result.fail("删除任务在ZooKeeper节点信息失败");
            }
        }
        //DB删除
        taskDtoMapper.deleteTask(oldTaskId.getId());//uk id=命名空间+taskid
        ZooKeeperTaskConfigurationTransferUtil.updateTaskUpdateVersion(curator);
        return Result.success();
    }

    /**
     * 删除zk上的任务结点。先持久化config/toDelete结点，让executor收到该事件，shutdown自身的该任务。如果所有executor都已经shutdown该任务，则才可以安全删除任务结点。
     *
     * @return 等待executor shutdown任务，等待一定时间后，如果executor还没完全shutdown，则放弃等待，返回false。
     * 否则，在等待时间内，executor都shutdown完全，则删除任务结点，并返回true。
     */
    private boolean removeTaskFromZooKeeper(Curator curator, String taskId) throws Exception {
        //首先设置暂停调度
        curator.merge(TaskNodePaths.getTaskBaseInfoNodePath(taskId, TaskBaseInfoNode.status), YN.N);
        //一旦打上删除标签 那么则不能进行启用。只能尝试禁用任务和删除。避免ZK节点信息不全。造成调度过程中出现各种任务。
        String deleteNodePath = TaskNodePaths.getTaskBaseInfoNodePath(taskId, "delete");
        if (!curator.checkExists(deleteNodePath)) {
            curator.create(deleteNodePath);
        }
        // 1.任务的executor全online的情况，添加toDelete节点，触发监听器动态删除节点
        String toDeleteNodePath = TaskNodePaths.getTaskBaseInfoNodePath(taskId, "toDelete");
        if (curator.checkExists(toDeleteNodePath)) {
            curator.delete(toDeleteNodePath);
        }
        curator.create(toDeleteNodePath);
        TimeUnit.SECONDS.sleep(10);//休眠10秒(一般情况不会删除任务的)
        //
        for (int i = 0; i < 20; i++) {
            // 2.任务的executor全offline的情况，或有几个online，几个offline的情况
            String executorsNodePath = TaskNodePaths.getTaskExecutorsNodePath(taskId);
            if (!curator.checkExists(executorsNodePath)) {
                // (1)如果不存在$Job/taskId/servers节点，说明该任务没有任何executor接管，可直接删除任务节点
                curator.delete(TaskNodePaths.getTaskNodePath(taskId));
                return true;
            }
            // (2)如果该任务servers下没有任何executor，可直接删除任务节点
            List<String> executorList = curator.getChildren(executorsNodePath);
            if (CollectionUtils.isEmpty(executorList)) {
                curator.delete(TaskNodePaths.getTaskNodePath(taskId));
                return true;
            }
            // (3)只要该任务没有一个能运行的该任务的executor在线，那么直接删除任务节点
            boolean hasOnlineExecutor = false;
            for (String executor : executorList) {
                if (curator.checkExists(ExecutorNodePaths.getExecutorSubNodePath(executor, "ip")) && curator.checkExists(TaskNodePaths.getExecutorStatus(taskId, executor))) {
                    hasOnlineExecutor = true;
                } else {
                    curator.delete(TaskNodePaths.getTaskExecutorNodePath(taskId, executor));
                }
            }
            if (!hasOnlineExecutor) {
                curator.delete(TaskNodePaths.getTaskNodePath(taskId));
                curator.delete("enableNode");//task配置信息外部节点
                return true;
            }
            try {
                Thread.sleep(1000);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }

        return false;

    }


    @Transactional(rollbackFor = Exception.class)
    @Override
    public Result enableTask(String nameSpace, String taskId, String userId) throws RuntimeException {
        TaskDto taskDto = taskDtoMapper.getTaskDtoByNamespaceKeyAndTaskId(nameSpace, taskId);
        if (taskDto == null) {
            throw new RuntimeException("不能启用该任务（" + taskId + "），因为该任务不存在");
        }//任务只能禁用不会真正的删除
        CuratorInfo curatorInfo = zooKeeperService.getCuratorInfo(nameSpace);
        Curator curator = CuratorFactory.getCurator(curatorInfo, CuratorFactory.shortTime_true);
        String enabledNodePath = TaskNodePaths.getTaskBaseInfoNodePath(taskId, TaskBaseInfoNode.status);
        String enabled = curator.getData(enabledNodePath);
        if (YN.Y.equals(taskDto.getStatus())) {
            String deleteNodePath = TaskNodePaths.getTaskBaseInfoNodePath(taskId, "delete");
            if (curator.checkExists(deleteNodePath)) {
                return Result.fail(String.format("当前任务处理删除阶段,不能启用该任务(%s)", taskId));
            }
            //throw new RuntimeException("该任务（" + taskId + "）已经处于启用状态");
            if (YN.Y.equals(enabled)) {
                throw new RuntimeException("该任务（" + taskId + "）已经处于启用状态");
            } else {
                //zk启用
                enableTask(curator, taskId);
                //enableTask(taskConfigurationDto, userId);
                ZooKeeperTaskConfigurationTransferUtil.updateTaskUpdateVersion(curator);
                return Result.success("启用任务成功");
            }
        } else {
            if (YN.Y.equals(enabled)) {
                //enableTask(zooKeeper, taskId);
                enableTask(taskDto, userId);
                ZooKeeperTaskConfigurationTransferUtil.updateTaskUpdateVersion(curator);
                return Result.success("启用任务成功");
            } else {
                //zk启用
                enableTask(curator, taskId);
                enableTask(taskDto, userId);
                ZooKeeperTaskConfigurationTransferUtil.updateTaskUpdateVersion(curator);
                return Result.success("启用任务成功");
            }
        }
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public Result disableTask(String nameSpace, String taskId, String userId) throws RuntimeException {
        TaskDto taskId1 = taskDtoMapper.getTaskDtoByNamespaceKeyAndTaskId(nameSpace, taskId);
        if (taskId1 == null) {
            throw new RuntimeException("不能禁用该任务（" + taskId + "），因为该任务不存在");
        }
        CuratorInfo curatorInfo = zooKeeperService.getCuratorInfo(nameSpace);
        Curator curator = CuratorFactory.getCurator(curatorInfo, CuratorFactory.shortTime_true);
        String enabledNodePath = TaskNodePaths.getTaskBaseInfoNodePath(taskId, TaskBaseInfoNode.status);
        String enabled = curator.getData(enabledNodePath);
        if (YN.N.equals(taskId1.getStatus())) {
            if (YN.N.equals(enabled)) {
                throw new RuntimeException("该任务（" + taskId + "）已经处于禁用状态");
            } else {
                disableTask(curator, taskId);
                //disableTask(taskConfigurationDto, userId);
                ZooKeeperTaskConfigurationTransferUtil.updateTaskUpdateVersion(curator);
                return Result.success("禁用任务成功");
            }
        } else {
            if (YN.N.equals(enabled)) {
                //disableTask(zooKeeper, taskId);
                disableTask(taskId1, userId);
                ZooKeeperTaskConfigurationTransferUtil.updateTaskUpdateVersion(curator);
                return Result.success("禁用任务成功");
            } else {
                disableTask(curator, taskId);
                disableTask(taskId1, userId);
                ZooKeeperTaskConfigurationTransferUtil.updateTaskUpdateVersion(curator);
                return Result.success("禁用任务成功");
            }

        }
    }

    private void enableTask(TaskDto taskDto, String userId) {
        taskDto.setStatus(YN.Y);
        taskDto.setLastUpdateBy(userId);
        taskDto.setLastUpdateTime(new Date());
        taskDtoMapper.updateTask(taskDto);
    }

    private void disableTask(TaskDto taskDto, String userId) {
        taskDto.setStatus(YN.N);
        taskDto.setLastUpdateBy(userId);
        taskDto.setLastUpdateTime(new Date());
        taskDtoMapper.updateTask(taskDto);
    }

    private void enableTask(Curator curator, String taskId) {
        String enableTaskNodePath = EnabledTaskNodePaths.getEnableTaskNodePath(taskId);
        boolean checkExists = curator.checkExists(enableTaskNodePath);
        if (!checkExists) {
            curator.merge(enableTaskNodePath, "");
        }

    }

    private void disableTask(Curator curator, String taskId) {
        String enableTaskNodePath = EnabledTaskNodePaths.getEnableTaskNodePath(taskId);
        boolean checkExists = curator.checkExists(enableTaskNodePath);
        if (checkExists) {
            curator.delete(enableTaskNodePath);
        }

    }

    @Override
    public Result stopTaskInstanceId(String nameSpace, String taskId, String taskInstanceId, final JSONObject triggeredData) throws RuntimeException {
        //暂时某个任务实例进行跑的场景有三种:
        // 1任务暂停以后指定某个时间后再跑，
        // 2任务暂停后，不指定时间重跑，可以手动指定重跑。
        // 3任务暂停，未来也不进行重跑，这种情况可能容易出现问题。故如果选择3则需要进行提示是否要这样做。
        // 为了简单:默认推荐方式为2【任务暂停后，不指定时间重跑，可以手动指定重跑。】，本框架只实现了该种暂停方式，在生成环境这种方式能够兼容其他两种场景。
        //
        List<String> executorIdList = getExecutorIdList(nameSpace, taskId);
        if (!(executorIdList != null && !executorIdList.isEmpty())) {
            throw new RuntimeException(String.format("没有executor接管该任务(%s)，不能立即终止", taskId));
        } else {
            CuratorInfo curatorInfo = zooKeeperService.getCuratorInfo(nameSpace);
            Curator curator = CuratorFactory.getCurator(curatorInfo, CuratorFactory.shortTime_true);
            String path = TaskInstanceManageNodePaths.getTaskInstanceStopNodePath(taskId, taskInstanceId);
            if (curator.checkExists(path)) {
                curator.delete(path);
            }
            String triggeredDataStr = JSON.toJSONString(triggeredData);
            curator.create(path, triggeredDataStr);
        }
        return Result.success();
    }


    @Override
    public Result rerunTaskInstanceId(final String nameSpace, final String taskId, String taskInstanceId, final JSONObject triggeredData) throws RuntimeException {
        CuratorInfo curatorInfo = zooKeeperService.getCuratorInfo(nameSpace);
        Curator curator = CuratorFactory.getCurator(curatorInfo, CuratorFactory.shortTime_true);
        List<TaskExecutorStatusInfo> taskExecutorStatusInfoList = getTaskExecutorStatusInfoList(nameSpace, taskId);
        if (CollectionUtils.isEmpty(taskExecutorStatusInfoList)) {
            throw new RuntimeException(String.format("没有executor接管该任务(%s)，不能立即启动", taskId));
        }
        boolean hasOnlineExecutor = false;
        for (TaskExecutorStatusInfo taskExecutorStatusInfo : taskExecutorStatusInfoList) {
            if (TaskExecutorStatus.ONLINE.equals(taskExecutorStatusInfo.getStatus())) {
                hasOnlineExecutor = true;
                log.info("runAtOnce nameSpace:{}, taskId:{}, executorName:{}, triggeredData:{}", nameSpace, taskId);
            }
        }

        if (!hasOnlineExecutor) {
            throw new RuntimeException(String.format("没有在线的executor接管该任务(%s)，不能立即启动", taskId));
        }
        //重跑 是否是新建实例，如果之前的实例存在那么则不能进行重跑，需要等待停止完毕才能进行重跑。
        //如果type=new 则新建实例。需要关注任务的捞取数据范围。
        //如果type=pause,需要指定之前的任务实例id,如果该实例id正在跑 那么不允许重跑。
        String type = triggeredData.getString("type");
        if ("new".equals(type)) {

        } else if ("resume".equals(type)) {

        } else {
            return Result.fail("暂时不支持重跑类型");
        }
        String path = TaskInstanceManageNodePaths.getTaskInstanceRestartNodePath(taskId, taskInstanceId);
        if (curator.checkExists(path)) {
            curator.delete(path);
        }
        String triggeredDataStr = JSON.toJSONString(triggeredData);
        curator.create(path, triggeredDataStr);
        return null;
    }

    @Override
    public Result runTaskInstanceId(String nameSpace, String taskId, final JSONObject triggeredData) throws RuntimeException {
        return null;
    }

    @Override
    public List<String> getTaskIdList(String nameSpace) throws RuntimeException {
        List<String> taskIdList = taskDtoMapper.getTaskIdListByNamespaceKey(nameSpace);
        return taskIdList != null ? taskIdList : new ArrayList<>();
    }

    @Override
    public List<TaskDto> getTaskDtoList(String nameSpace) throws RuntimeException {

        List<TaskDto> taskDtoList = taskDtoMapper.getTaskDtoListByNamespaceKey(nameSpace);
        return taskDtoList;
    }

    @Override
    public TaskDto getTaskConfiguration(String nameSpace, String taskId) throws RuntimeException {
        TaskDto taskDto = taskDtoMapper.getTaskDtoByNamespaceKeyAndTaskId(nameSpace, taskId);
        if (taskDto == null) {
            throw new RuntimeException(String.format("该任务(%s)不存在", taskId));
        }
        return taskDto;
    }

    @Override
    public List<String> getTaskIdListFromZooKeeper(String nameSpace) throws RuntimeException {
        CuratorInfo curatorInfo = zooKeeperService.getCuratorInfo(nameSpace);
        Curator curator = CuratorFactory.getCurator(curatorInfo, CuratorFactory.shortTime_true);
        String tasksNodePath = TaskNodePaths.getROOT();
        List<String> taskIdList = curator.getChildren(tasksNodePath);
        if (taskIdList == null) {
            return new ArrayList<>();
        }
        List<String> $taskIdList = new ArrayList<>();
        for (String taskId : taskIdList) {
            // 如果config节点存在才视为正常任务，其他异常任务在其他功能操作时也忽略
            if (curator.checkExists(TaskNodePaths.getTaskBaseInfoNodePath(taskId))) {
                $taskIdList.add(taskId);
            }
        }
        Collections.sort($taskIdList);
        return $taskIdList;
    }

    @Override
    public List<TaskDto> getTaskInfoList(final String nameSpace) throws RuntimeException {

        List<TaskDto> taskDtoList = getTaskDtoList(nameSpace);
        List<TaskDto> taskConfigurationList = new ArrayList<>();
        if (taskDtoList != null) {
            for (TaskDto taskDto : taskDtoList) {
                try {
                    TaskDto taskConfiguration1 = getTaskInfo(nameSpace, taskDto.getTaskId());
                    taskConfigurationList.add(taskConfiguration1);
                } catch (Exception e) {
                    log.error("getRestApiJobInfos exception:", e);
                    continue;
                }
            }
        }
        return taskConfigurationList;
    }


    @Override
    public TaskDto getTaskInfo(String nameSpace, final String taskId) throws RuntimeException {
        CuratorInfo curatorInfo = zooKeeperService.getCuratorInfo(nameSpace);
        Curator curator = CuratorFactory.getCurator(curatorInfo, CuratorFactory.shortTime_true);
        TaskDto taskConfiguration = ZooKeeperTaskConfigurationTransferUtil.getTaskConfigurationFromZooKeeper(curator, taskId);
        return taskConfiguration;
    }


    @Override
    public List<TaskItemExecuteInfo> getTaskItemExecuteInfoList(String nameSpace, String taskId) throws Exception {
        CuratorInfo curatorInfo = zooKeeperService.getCuratorInfo(nameSpace);
        Curator curator = CuratorFactory.getCurator(curatorInfo, CuratorFactory.shortTime_true);
        TaskDto taskConfiguration = getTaskConfiguration(nameSpace, taskId);
        if (!taskConfiguration.getStatus().equals("Y")) {
            return new ArrayList<>();
        }
        // 如果execution节点不存在则返回空List
        List<TaskItemRenewalNodeInfo> list = RenewalNodePaths.getTaskItemRenewalNodeInfoList(curator, taskId);
        List<TaskItemExecuteInfo> taskItemExecuteInfoList = new ArrayList<>();
        for (TaskItemRenewalNodeInfo taskItemIndexRenewalNodeInfo : list) {
            TaskItemExecuteInfo taskItemExecuteInfo = getTaskItemExecuteInfo(curator, taskItemIndexRenewalNodeInfo);
            taskItemExecuteInfoList.add(taskItemExecuteInfo);
        }
        Collections.sort(taskItemExecuteInfoList);

        return taskItemExecuteInfoList;
    }


    private TaskItemExecuteInfo getTaskItemExecuteInfo(Curator curator, TaskItemRenewalNodeInfo taskItemIndexRenewalNodeInfo) {
        TaskItemExecuteInfo taskItemExecuteInfo = new TaskItemExecuteInfo();
        taskItemExecuteInfo.setTaskId(taskItemIndexRenewalNodeInfo.getTaskId());
        taskItemExecuteInfo.setTaskInstanceId(taskItemIndexRenewalNodeInfo.getTaskInstanceId());
        taskItemExecuteInfo.setTaskStage(taskItemIndexRenewalNodeInfo.getTaskStage());
        taskItemExecuteInfo.setTaskStage$type(taskItemExecuteInfo.getTaskStage$type());
        taskItemExecuteInfo.setTaskStage$type$index(taskItemIndexRenewalNodeInfo.getTaskStage$type$index());
        TaskItemExecuteRenewalInfo taskItemExecuteRenewalInfo = RenewalNodePaths.getTaskItemExecuteRenewalInfo(curator, taskItemIndexRenewalNodeInfo);
        if (taskItemExecuteRenewalInfo == null) {
            taskItemExecuteInfo.setStatus(TaskItemExecuteStatus.NULL);
            return taskItemExecuteInfo;
        }
        if (TaskItemExecuteStatus.PENDING == taskItemExecuteRenewalInfo.getStatus()) {
            taskItemExecuteInfo.setStatus(TaskItemExecuteStatus.PENDING);
            return taskItemExecuteInfo;
        }
        taskItemExecuteInfo.setExecutorId(taskItemExecuteRenewalInfo.getExecutorId());
        taskItemExecuteInfo.setBeginTime(taskItemExecuteRenewalInfo.getBeginTime());
        taskItemExecuteInfo.setReportTime(taskItemExecuteRenewalInfo.getReportTime());
        taskItemExecuteInfo.setTimeConsumed((taskItemExecuteRenewalInfo.getReportTime() - taskItemExecuteRenewalInfo.getBeginTime()) / 1000);
        //
        if (TaskItemExecuteStatus.COMPLETED == taskItemExecuteRenewalInfo.getStatus()) {
            taskItemExecuteInfo.setStatus(TaskItemExecuteStatus.COMPLETED);
            return taskItemExecuteInfo;
        }
        if (TaskItemExecuteStatus.FAILED == taskItemExecuteRenewalInfo.getStatus()) {
            taskItemExecuteInfo.setStatus(TaskItemExecuteStatus.FAILED);
            return taskItemExecuteInfo;
        }
        if (TaskItemExecuteStatus.TIMEOUT == taskItemExecuteRenewalInfo.getStatus()) {
            taskItemExecuteInfo.setStatus(TaskItemExecuteStatus.TIMEOUT);
            return taskItemExecuteInfo;
        }
        if (TaskItemExecuteStatus.RUNNING == taskItemExecuteRenewalInfo.getStatus()) {
            taskItemExecuteInfo.setStatus(TaskItemExecuteStatus.RUNNING);
            return taskItemExecuteInfo;
        }
        throw new IllegalStateException("暂时还不支持其他的状态");
    }


    /////////////////////////////////////////////////////////////////////////////////
    @Override
    public List<Executor> getExecutorList(String nameSpace, String taskId) throws RuntimeException {
        TaskDto taskDto = taskDtoMapper.getTaskDtoByNamespaceKeyAndTaskId(nameSpace, taskId);
        if (taskDto == null) {
            throw new RuntimeException("不能获取该任务（" + taskId + "）可选择的优先Executor，因为该任务不存在");
        }
        CuratorInfo curatorInfo = zooKeeperService.getCuratorInfo(nameSpace);
        Curator curator = CuratorFactory.getCurator(curatorInfo, CuratorFactory.shortTime_true);
        String taskNodePath = TaskNodePaths.getTaskNodePath(taskId);
        if (!curator.checkExists(taskNodePath)) {
            return null;
        }
        List<Executor> executorList = new ArrayList<>();
        String executorsNodePath = ExecutorNodePaths.getRoot();
        if (!curator.checkExists(executorsNodePath)) {
            return executorList;
        }
        List<String> executorIdList = curator.getChildren(executorsNodePath);
        if (executorIdList == null) {
            executorIdList = new ArrayList<>();
        }
        if (!executorIdList.isEmpty()) {
            for (String executorId : executorIdList) {
                if (!executorId.startsWith("@")) {
                    Executor executor = new Executor();
                    executor.setExecutorType(ExecutorType.PHYSICAL_MACHINE);
                    executor.setExecutorId(executorId);
                    executor.setOwnTraffic(curator.checkExists(ExecutorNodePaths.getExecutorOwnTrafficNodePath(executorId)));
                    String ip = curator.getData(ExecutorNodePaths.getExecutorIpNodePath(executorId));
                    if (StringUtils.isNotBlank(ip)) {
                        executor.setExecutorStatus(ExecutorStatus.ONLINE);
                        executor.setIp(ip);
                    } else {
                        executor.setExecutorStatus(ExecutorStatus.OFFLINE);
                    }
                    executorList.add(executor);
                } else {
                    Executor executor = new Executor();
                    executor.setExecutorType(ExecutorType.CONTAINER);
                    executor.setExecutorId(executorId);
                    executor.setOwnTraffic(curator.checkExists(ExecutorNodePaths.getExecutorOwnTrafficNodePath(executorId)));
                    String ip = curator.getData(ExecutorNodePaths.getExecutorIpNodePath(executorId));
                    if (StringUtils.isNotBlank(ip)) {
                        executor.setExecutorStatus(ExecutorStatus.ONLINE);
                        executor.setIp(ip);
                    } else {
                        executor.setExecutorStatus(ExecutorStatus.OFFLINE);
                    }
                    executorList.add(executor);
                }

            }
        }
        String preferExecutorIdListNodePath = TaskNodePaths.getTaskBaseInfoNodePath(taskId, TaskNodePaths.prefer_executor_id_list);
        if (!curator.checkExists(preferExecutorIdListNodePath)) {
            return executorList;
        }
        String preferExecutorIdListSting = curator.getData(preferExecutorIdListNodePath);
        if (Strings.isNullOrEmpty(preferExecutorIdListSting)) {
            return executorList;
        }
        String[] preferExecutorIdList = preferExecutorIdListSting.split(Symbols.COMMA);
        for (String preferExecutorId : preferExecutorIdList) {
            if (!preferExecutorId.startsWith("@")) {
                boolean include = false;
                for (Executor executor : executorList) {
                    if (executor.getExecutorId().equals(preferExecutorId)) {
                        include = true;
                        break;
                    }
                }
                if (!include) {
                    Executor executor = new Executor();
                    executor.setExecutorId(preferExecutorId);
                    executor.setExecutorType(ExecutorType.PHYSICAL_MACHINE);
                    executor.setExecutorStatus(ExecutorStatus.DELETED);
                    executorList.add(executor);
                }
            } else {
                boolean include = false;
                for (Executor executor : executorList) {
                    if (executor.getExecutorId().equals(preferExecutorId)) {
                        include = true;
                        break;
                    }
                }
                if (!include) {
                    Executor executor = new Executor();
                    executor.setExecutorId(preferExecutorId);
                    executor.setExecutorType(ExecutorType.CONTAINER);
                    executor.setExecutorStatus(ExecutorStatus.DELETED);
                    executorList.add(executor);
                }
            }
        }
        return executorList;
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void setPreferExecutorIdList(String nameSpace, String taskId, String preferExecutorIdListSting) throws RuntimeException {
        TaskDto taskDto = taskDtoMapper.getTaskDtoByNamespaceKeyAndTaskId(nameSpace, taskId);
        if (taskDto == null) {
            throw new RuntimeException("设置该任务（" + taskId + "）优先Executor失败，因为该任务不存在");
        }
        taskDto.setPreferExecutorIdList(preferExecutorIdListSting);
        taskDtoMapper.updatePreferExecutorIdList(taskDto);
        // save to zk
        CuratorInfo curatorInfo = zooKeeperService.getCuratorInfo(nameSpace);
        Curator curator = CuratorFactory.getCurator(curatorInfo, CuratorFactory.shortTime_true);
        String taskConfigPreferListNodePath = TaskNodePaths.getTaskBaseInfoNodePath(taskId, TaskNodePaths.prefer_executor_id_list);
        curator.merge(taskConfigPreferListNodePath, preferExecutorIdListSting);

        //

    }


    @Override
    public List<TaskExecutor> getTaskExecutorList(String nameSpace, String taskId) throws RuntimeException {
        CuratorInfo curatorInfo = zooKeeperService.getCuratorInfo(nameSpace);
        Curator curator = CuratorFactory.getCurator(curatorInfo, CuratorFactory.shortTime_true);
        String executorsNodePath = TaskNodePaths.getTaskExecutorsNodePath(taskId);
        List<String> executorIdList = curator.getChildren(executorsNodePath);
        List<TaskExecutor> taskExecutorList = new ArrayList<>();
        if (executorIdList != null && !executorIdList.isEmpty()) {
            for (String executorId : executorIdList) {
                TaskExecutor taskExecutor = getTaskExecutor(curator, taskId, executorId);
                taskExecutorList.add(taskExecutor);
            }
        }
        return taskExecutorList;
    }

    private TaskExecutor getTaskExecutor(Curator curator, String taskId, String executorId) {
        TaskExecutor taskExecutor = new TaskExecutor();
        taskExecutor.setTaskId(taskId);
        taskExecutor.setExecutorId(executorId);
        taskExecutor.setContainer(executorId.startsWith("@"));
        taskExecutor.setIp(curator.getData(TaskNodePaths.getTaskExecutorNodePath(taskId, executorId, TaskExecutorNode.ip)));
        taskExecutor.setStatus(getTaskExecutorStatus(curator, taskId, executorId));
        //
        taskExecutor.setTaskItemAssign(curator.getData(TaskNodePaths.getTaskExecutorNodePath(taskId, executorId, TaskExecutorNode.TASK_ITEM_ASSIGN)));
        //
        String processSuccessCount = curator.getData(TaskNodePaths.getTaskExecutorNodePath(taskId, executorId, TaskExecutorNode.processSuccessCount));
        taskExecutor.setProcessSuccessCount(null == processSuccessCount ? 0 : Integer.parseInt(processSuccessCount));
        String processFailureCount = curator.getData(TaskNodePaths.getTaskExecutorNodePath(taskId, executorId, TaskExecutorNode.processFailureCount));
        taskExecutor.setProcessFailureCount(null == processFailureCount ? 0 : Integer.parseInt(processFailureCount));
        //
        return taskExecutor;
    }

    @Override
    public List<TaskExecutorStatusInfo> getTaskExecutorStatusInfoList(String nameSpace, String taskId) throws RuntimeException {
        CuratorInfo curatorInfo = zooKeeperService.getCuratorInfo(nameSpace);
        Curator curator = CuratorFactory.getCurator(curatorInfo, CuratorFactory.shortTime_true);
        List<TaskExecutorStatusInfo> taskExecutorStatusInfoList = new ArrayList<>();
        List<String> executorIdList = getExecutorIdList(nameSpace, taskId);
        if (executorIdList != null) {
            for (String executorId : executorIdList) {
                taskExecutorStatusInfoList.add(getTaskExecutorStatusInfo(curator, taskId, executorId));
            }
        }
        return taskExecutorStatusInfoList;
    }

    @Override
    public List<String> getExecutorIdList(String nameSpace, String taskId) throws RuntimeException {
        CuratorInfo curatorInfo = zooKeeperService.getCuratorInfo(nameSpace);
        Curator curator = CuratorFactory.getCurator(curatorInfo, CuratorFactory.shortTime_true);
        String taskNodePath = TaskNodePaths.getTaskNodePath(taskId);
        if (!curator.checkExists(taskNodePath)) {
            return null;
        }
        String executorsNodePath = TaskNodePaths.getTaskExecutorsNodePath(taskId);
        List<String> executorIdList = curator.getChildren(executorsNodePath);
        if (executorIdList == null || CollectionUtils.isEmpty(executorIdList)) {
            return new ArrayList<>();
        }
        return executorIdList;
    }

    private TaskExecutorStatusInfo getTaskExecutorStatusInfo(Curator curator, String taskId, String executorId) {
        TaskExecutorStatusInfo taskExecutorStatus = new TaskExecutorStatusInfo();
        taskExecutorStatus.setTaskId(taskId);
        taskExecutorStatus.setExecutorId(executorId);
        taskExecutorStatus.setStatus(getTaskExecutorStatus(curator, taskId, executorId));
        return taskExecutorStatus;
    }

    private String getTaskExecutorStatus(Curator curator, String taskId, String executorId) {
        return curator.getData(TaskNodePaths.getTaskExecutorNodePath(taskId, executorId, TaskExecutorNode.status));
    }


}

